package geektime.spring.data.reactive.redisdemo;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class DemoSubsriber implements Subscriber<Integer> {

  static final Logger LOGGER = LoggerFactory.getLogger(DemoSubsriber.class);

  @Override
  public void onSubscribe(Subscription s) {
    LOGGER.info("request max");
    s.request(Integer.MAX_VALUE);
  }

  @Override
  public void onNext(Integer integer) {
    LOGGER.info("get data:{}", integer);

  }

  @Override
  public void onError(Throwable t) {

  }

  @Override
  public void onComplete() {
    LOGGER.info("onComplete");
  }


  @Test
  public void testDoOnMethods() {
    Flux.range(1, 3)
//                .log()
        .doOnSubscribe(e -> LOGGER.info("doOnSubscribe:{}", e))
        .doOnEach(e -> LOGGER.info("doOnEach:{}", e))
        .doOnError(e -> LOGGER.info("doOnError:{}", e))
        .doOnNext(e -> LOGGER.info("doOnNext:{}", e))
        .doOnRequest(e -> LOGGER.info("doOnRequest:{}", e))
        .doOnTerminate(new Runnable() {
          @Override
          public void run() {
            LOGGER.info("doOnTerminate");
          }
        })
        .doOnCancel(new Runnable() {
          @Override
//                    public void run() {
//                    LOGGER.info("doOnCancel");
//                  }
          public void run() {
            LOGGER.info("操作被取消");
          }
        })
        .doOnComplete(new Runnable() {
          @Override
          public void run() {
            LOGGER.info("doOnComplete");
          }
        })
        .subscribe(new DemoSubsriber());

  }

  @Test
  public void testDoOnMethods1() throws InterruptedException {
    Flux.range(1,3).publishOn(Schedulers.single())
        .doOnSubscribe(e -> LOGGER.info("doOnSubscribe:{}", e))
        .doOnComplete(() -> LOGGER.info("doOnComplete"))
        .subscribe(c -> LOGGER.info("subscribe-->Insert{} records", c), e -> LOGGER.info("subscribe-->doOnSubscribe:{}", e),
            new Runnable() {
              @Override
              public void run() {
                LOGGER.info("subscribe-->doOnTerminate");
              }
            });
Thread.sleep(900000L);

  }

//  @Test
//  public void testConnectableFluxRefConnect() throws InterruptedException {
//
//    Flux<Long> source = Flux.interval(Duration.ofMillis(500))
//        .doOnSubscribe(s -> System.out.println("上遊收到訂閱"))
//        .doOnCancel(() -> System.out.println("上遊發布者斷開連接doOnCancel"));
//
//    Flux<Long> refCounted = source.publish().refCount(2, Duration.ofSeconds(2));
//
//    System.out.println("第一個訂閱者訂閱");
//    Disposable sub1 = refCounted.subscribe(l -> System.out.println("sub1: " + l));
//
//    TimeUnit.SECONDS.sleep(1);
//    System.out.println("第二個訂閱者訂閱");
//    Disposable sub2 = refCounted.subscribe(l -> System.out.println("sub2: " + l));
//
//    TimeUnit.SECONDS.sleep(1);
//    System.out.println("第一個訂閱者取消訂閱");
//    sub1.dispose();
//
//    TimeUnit.SECONDS.sleep(1);
//    System.out.println("第二個訂閱者取消訂閱");
//    sub2.dispose();
//
//    TimeUnit.SECONDS.sleep(1);
//    System.out.println("第三個訂閱者訂閱");
//    Disposable sub3 = refCounted.subscribe(l -> System.out.println("sub3: " + l));
//
//    TimeUnit.SECONDS.sleep(1);
//    System.out.println("第三個訂閱者取消訂閱");
//    sub3.dispose();
//
//    TimeUnit.SECONDS.sleep(3);
//    System.out.println("第四個訂閱者訂閱");
//    Disposable sub4 = refCounted.subscribe(l -> System.out.println("sub4: " + l));
//    TimeUnit.SECONDS.sleep(1);
//    System.out.println("第五個訂閱者訂閱");
//    Disposable sub5 = refCounted.subscribe(l -> System.out.println("sub5: " + l));
//    TimeUnit.SECONDS.sleep(2);
//  }
}